package com.xxxx;

import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class Hello03FlinkKeyedState {
    public static void main(String[] args) throws Exception {
        //1-环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //全局并行度设置
        env.setParallelism(1);

        //2-数据源source
        DataStreamSource<Tuple3<String, String, Long>> tupleStream = env.fromElements(
                Tuple3.of("上海", "普陀区", 111L), Tuple3.of("上海", "徐汇区", 222L),
                Tuple3.of("北京", "西城区", 1333L), Tuple3.of("北京", "海淀区", 444L),
                Tuple3.of("上海", "杨浦区", 555L), Tuple3.of("上海", "浦东新区", 2666L),
                Tuple3.of("北京", "东城区", 777L), Tuple3.of("上海", "黄浦区", 888L)
        );

        //3-数据的transformation
        //SingleOutputStreamOperator<Tuple3<String, String, Long>> maxDataStream = tupleStream
        //        .keyBy(0)
        //        .maxBy(2);

        SingleOutputStreamOperator<String> stateDataStream = tupleStream
                .keyBy(0) // 城市city分组
                .map(new RichMapFunction<Tuple3<String, String, Long>, String>() {
                    // step1. 定义存储转态数据结构
                    // transient是类型修饰符，只能用来修饰字段。在对象序列化的过程中，标记为transient的变量不会被序列化
                    private transient ValueState<Long> valueState = null;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        // step2. 初始化状态的值
                        valueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<Long>("maxState", Long.class)
                        );
                    }

                    @Override
                    public String map(Tuple3<String, String, Long> tuple) throws Exception {
					/*
						Key: 上海，  tuple： 上海,普陀区,488
					 */
                        // step3. 从以前状态中获取值
                        Long historyValue = valueState.value();

                        // 获取当前key中的值
                        Long currentValue = tuple.f2;

                        // 判断历史值是否为null，如果key的数据第1次出现，以前没有状态
                        if (null == historyValue || currentValue > historyValue) {
                            // step4. 更新状态值
                            valueState.update(currentValue);
                        }

                        // 返回处理结果
                        return tuple.f0 + ", " + valueState.value();
                    }
                });
        stateDataStream.printToErr();

        //5-获取执行的execute
        env.execute("Hello03FlinkKeyedState" + System.currentTimeMillis());

    }
}
